package com.bytes32.stats;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Iterator;

/**
 * A type of quantile that is build by merging 
 * from previous quantiles into one final solution
 * 
 * @author murariuf
 *
 */
public class ReduceQuantiles extends LargeQuantiles{

	public ReduceQuantiles(Config conf) {
		super(Integer.MAX_VALUE, conf); /* sampling is disabled when building the tree */
		this.current = null; /* no assigned current buffer for this quantile */
		this.buffers.clear();
	}

	/**
	 * Not supported for ReduceQuantiles 
	 */
	@Override
	public void add(double... input) {
		throw new UnsupportedOperationException("Reduce Quantiles are built from other quantiles, use LargeQuantiles instead ");
	}
	
	
	public ReduceQuantiles reduce(Iterator<LargeQuantiles> finalOutputs){
		while (finalOutputs.hasNext()){
			Collection<QBuffer> pair = finalOutputs.next().buffers();
			if (pair.size() > 2){
				/* something did not run in the pre-reduce phase */
				pair = MergeAndSelect.preReduce(pair);
			}
			/* for now disable sampling */
			this.setR(1);
			for (QBuffer buffer : pair){
				if (buffer.status() == BufferStatus.FULL){
					/* deal with full buffer, simply add it to the list, if there are too many collapse them  */
					if (this.buffers.size() >= this.getB()){
						Collection<QBuffer> toCollapse = new ArrayList<QBuffer>(this.buffers.size()-1);/* avoid the last one */
						for (QBuffer buf : this.buffers) if (buf.status() == BufferStatus.FULL) toCollapse.add(buf); /* can I get away with this? :)*/
						this.buffers.clear(); 
						this.buffers.add(MergeAndSelect.collapse(toCollapse, 0));
						this.buffers.add(this.current);
					} else {
						this.buffers.addFirst(buffer);
					}
				} else if (buffer.status() == BufferStatus.PARTIAL){
					if (this.current == null){
						this.current = buffer;
						this.buffers.add(this.current);
						continue; /* nothing else to do */
					}
					if (this.current.weight() == buffer.weight()){
						this._add(Arrays.copyOf(buffer.toArray(), buffer.getLength()));
					} else {
						/* the one with the largest weight becomes the new current partial buffer  */
						this.current = this.current.weight()<buffer.weight()?buffer:this.current;
						/* set the sampling rate to largest / smallest */
						setR(this.current.weight()<buffer.weight()?buffer.weight()/this.current.weight():this.current.weight()/buffer.weight());
						/* simply add the doubles of the buffer with the smallest weight */
						this._add(this.current.weight()<buffer.weight()?this.current.toArray():buffer.toArray());
						/* reset sampling to none */
						this.setR(1);
					}
				} else {
					continue; /* don't bother if the buffer is empty */
				}
			}
		}
		return this;
	}
	
}
